# Copyright (C) 2026. Logiciel élaboré par l'État, via le DEPS.
# Nom des auteures : Flore Cornuet et Camille Parent, DEPS.
# Ce programme informatique a été développé par le DEPS.
# Il permet de produire les résultats du Culture Études 1-2026.

# Ce programme a été exécuté en septembre 2026 avec la version 4.5.3 de R.
# La publication peut être consultée sur le site du Deps. 
# Ce programme utilise les données du pass Culture et du recensement de la population 2022.

# Bien qu'il n'existe aucune obligation légale à ce sujet, les utilisateurs de ce programme sont
# invités à signaler au DEPS leurs travaux issus de la réutilisation de ce code, ainsi que les
# éventuels problèmes ou anomalies qu'ils y rencontreraient, en écrivant à contact.deps@culture.gouv.fr .
# Ce logiciel est régi par la licence EUPL V1.2 soumise au droit français et européen, respectant
# les principes de diffusion des logiciels libres. Vous pouvez utiliser, modifier et/ou redistribuer
# ce programme sous les conditions de la licence EUPL V1.2, telle que diffusée sur
# https://eupl.eu/1.2/fr/ .
# En contrepartie de l'accessibilité au code source et des droits de copie, de modification et de
# redistribution accordés par cette licence, il n'est offert aux utilisateurs qu'une garantie
# limitée. Pour les mêmes raisons, seule une responsabilité restreinte pèse sur les auteures du
# programme, le titulaire des droits patrimoniaux et les concédants successifs.
# À cet égard l'attention de l'utilisateur est attirée sur les risques associés au chargement, à
# l'utilisation, à la modification et/ou au développement et à la reproduction du logiciel par
# l'utilisateur étant donné sa spécificité de logiciel libre, qui peut le rendre complexe à
# manipuler et qui le réserve donc à des développeurs et des professionnels avertis possédant des
# connaissances informatiques approfondies. Les utilisateurs sont donc invités à charger et tester
# l'adéquation du logiciel à leurs besoins dans des conditions permettant d'assurer la sécurité de
# leurs systèmes et ou de leurs données et, plus généralement, à l'utiliser et l'exploiter dans les
# mêmes conditions de sécurité.


# chargement des packages ####

library(arrow)
library(duckdb)
library(tictoc)
library(ggplot2)
library(stringr)
library(kableExtra)
library(dplyr)
library(data.table)
library(tidyr)
library(RColorBrewer)
library(dbplyr)
library(DBI)
library(glue)
library(readxl)
library(tidyverse)
library(lubridate)




# Connexion DuckDB ####

con <- dbConnect(duckdb())
dbExecute(con, "SET threads = 4")
dbExecute(con, "SET preserve_insertion_order = false")
dbExecute(con, "SET memory_limit = '16GB'")

# nom des colonnes ####

# colnames(open_dataset(paste0(chemin_donnees, "exp_ministry__offer_metadata/")))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__deposit/"
# )))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__booking/"
# )))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__offer/"
# )))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__venue/"
# )))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__venue_address/"
# )))
# 
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__offerer/"
# )))
# colnames(open_dataset(paste0(chemin_donnees, "offer_dict.parquet")))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__user_address/"
# )))
# colnames(open_dataset(paste0(
#   chemin_donnees, "exp_ministry__geo_iris/"
# )))

# _____________________________________________________________
# USERS_FILTRES ####
# Bénéficiaires du champ : users ayant au moins un deposit créé
# entre le 21/05/2021 et le 28/02/2025
# _____________________________________________________________

dbExecute(con, glue("
  CREATE OR REPLACE TABLE users_filtres AS
  SELECT DISTINCT user_id
  FROM read_parquet('{chemin_donnees}exp_ministry__deposit/*.parquet')
  WHERE deposit_creation_date >= '2021-05-21'
    AND deposit_creation_date <  '2025-02-28'
"))

dbExecute(con, glue("
  COPY users_filtres
  TO '{chemin_intermediaires}users_filtres.parquet'
  (FORMAT PARQUET)
"))

gc()


# DEPOSITS_FILTRES ####

dbExecute(con, glue("
  CREATE OR REPLACE TABLE deposits_filtres AS
  SELECT d.deposit_id, d.deposit_amount, d.user_id
  FROM read_parquet('{chemin_donnees}exp_ministry__deposit/*.parquet') d
  INNER JOIN users_filtres u ON d.user_id = u.user_id
  WHERE d.deposit_creation_date >= '2021-05-21'
    AND d.deposit_creation_date <  '2025-02-28'
"))

dbExecute(con, glue("
  COPY deposits_filtres
  TO '{chemin_intermediaires}deposits_filtres.parquet'
  (FORMAT PARQUET)
"))

gc()


# _____________________________________________________________
# USERS_FILTRES_INFO ####
# Informations détaillées sur les bénéficiaires du champ.
# Les recrédits postérieurs au 28/02/2025 sont exclus.
# _____________________________________________________________

dbExecute(con, glue("
  CREATE OR REPLACE TABLE users_filtres_info AS
  SELECT
    u.user_id,
    usr.user_birth_date,
    usr.user_civility,
    a.user_region_name,
    d.deposit_amount,
    d.total_recredit_amount,
    d.deposit_creation_date,
    d.deposit_update_date,
    d.total_recredit,
    d.deposit_id,

    -- Premier versement
    CASE
      WHEN d.deposit_amount = 80 THEN 20
      ELSE d.deposit_amount - d.total_recredit_amount
    END AS deposit_premier_versement,

    -- Deuxième versement
    CASE
      WHEN d.deposit_amount = 80                 THEN 30
      WHEN d.deposit_amount IN (50, 60)          THEN d.total_recredit_amount
      WHEN d.deposit_update_date <= '2025-02-28' THEN d.total_recredit_amount
      ELSE 0
    END AS deposit_deuxieme_versement,

    -- Troisième versement (uniquement pour les dépôts de 80€)
    CASE
      WHEN d.deposit_amount = 80 THEN 30
    END AS deposit_troisieme_versement,

    -- Dates des versements
    d.deposit_creation_date AS date_premier_versement,

    CASE
      WHEN d.deposit_amount = 80
        THEN CAST(CAST(usr.user_birth_date AS DATE) + INTERVAL '16' YEAR AS TIMESTAMP)
      ELSE d.deposit_update_date
    END AS date_deuxieme_versement,

    CASE
      WHEN d.deposit_amount = 80 THEN d.deposit_update_date
      ELSE NULL
    END AS date_troisieme_versement

  FROM users_filtres u
  INNER JOIN read_parquet('{chemin_donnees}exp_ministry__user/*.parquet')         usr USING (user_id)
  LEFT JOIN  read_parquet('{chemin_donnees}exp_ministry__user_address/*.parquet') a   USING (user_id)
  LEFT JOIN  read_parquet('{chemin_donnees}exp_ministry__deposit/*.parquet')      d
    ON u.user_id = d.user_id
    AND d.deposit_creation_date <= '2025-02-28'
"))

dbExecute(con, glue("
  COPY users_filtres_info
  TO '{chemin_intermediaires}users_filtres_info.parquet'
  (FORMAT PARQUET)
"))

gc()


# _____________________________________________________________
# BOOKINGS_FILTRES ####
# Réservations du champ : users du champ, créées avant le 28/02/2025,
# avec statut 'used' ou 'reimbursed' (hors annulations)
# _____________________________________________________________

dbExecute(con, glue("
  CREATE OR REPLACE TABLE offer_dict AS
  SELECT * FROM read_parquet('{chemin_donnees}offer_dict.parquet')
"))

dbExecute(con, glue("
  CREATE OR REPLACE TABLE bookings_filtres AS
  SELECT
    b.booking_id,
    b.user_id,
    b.offer_id,
    b.deposit_id,
    b.booking_intermediary_amount,
    b.booking_quantity,
    b.booking_intermediary_amount,
    o.offer_category
  FROM read_parquet('{chemin_donnees}exp_ministry__booking/*.parquet') b
  INNER JOIN users_filtres u USING (user_id)
  LEFT JOIN  offer_dict    o USING (offer_id)
  WHERE b.booking_creation_date <= '2025-02-28'
    AND b.booking_status IN ('reimbursed', 'used')
"))

dbExecute(con, "DROP TABLE offer_dict")

dbExecute(con, glue("
  COPY bookings_filtres
  TO '{chemin_intermediaires}bookings_filtres.parquet'
  (FORMAT PARQUET)
"))

gc()

colnames(open_dataset(paste0(
  chemin_intermediaires, "bookings_filtres.parquet"
)))

dbExecute(
  con,
  sprintf(
    "CREATE OR REPLACE TABLE bookings_filtres AS SELECT * FROM read_parquet('%s')",
    paste0(
      chemin_donnees,
      "tables_intermediaires/bookings_filtres.parquet"
    )
  )
)

# on ajoute deposit_amount 
dbExecute(con, glue("
  CREATE OR REPLACE TABLE deposit_tmp AS
  SELECT deposit_id, deposit_amount
  FROM read_parquet('{chemin_donnees}exp_ministry__deposit/*.parquet')
"))

# 2. Jointure beaucoup plus rapide
dbExecute(con, "
  CREATE OR REPLACE TABLE bookings_filtres AS
  SELECT b.*, d.deposit_amount
  FROM bookings_filtres b
  LEFT JOIN deposit_tmp d ON b.deposit_id = d.deposit_id
")

# on supprime deposit de con

dbExecute(con, "DROP TABLE deposit_tmp")

# on a tous les bookings used ou reimbursed, créés avant le 28 février 2025, des gens de notre champ.
# faisons maintenant le filtre sur l'expiration des deposit

dbExecute(con, glue("
  CREATE OR REPLACE TABLE deposit_filtres AS
  SELECT *
  FROM read_parquet('{chemin_donnees}exp_ministry__deposit/*.parquet')
  WHERE deposit_expiration_date < '2025-02-28'
"))

# on garde les lignes de booking_filtres quand le deposit_id est présent dans deposit_filtres

dbExecute(con, "
  CREATE OR REPLACE TABLE bookings_filtres AS
  SELECT b.*
  FROM bookings_filtres b
  INNER JOIN deposit_filtres d USING (deposit_id)
")

# donc on a :
# - les user de notre champ
# - les booking du chanmp
# -les deposit du champ

# on copie colle la nouvelle booking_filtres sur le disque

dbExecute(con, glue("
  COPY bookings_filtres
  TO '{chemin_intermediaires}bookings_filtres.parquet'
  (FORMAT PARQUET)
"))

# _____________________________________________________________
# BOOKINGS_FILTRES_META ####
# Bookings enrichies avec les métadonnées des offres
# _____________________________________________________________

dbExecute(con, glue("
  CREATE OR REPLACE TABLE bookings_filtres_meta AS
  SELECT b.*, m.* EXCLUDE (offer_id)
  FROM read_parquet('{chemin_intermediaires}bookings_filtres.parquet') b
  LEFT JOIN read_parquet('{chemin_donnees}exp_ministry__offer_metadata/*.parquet') m
    USING (offer_id)
"))

dbExecute(con, glue("
  COPY bookings_filtres_meta
  TO '{chemin_intermediaires}bookings_filtres_meta/'
  (FORMAT PARQUET, PARTITION_BY offer_category, OVERWRITE_OR_IGNORE TRUE)
"))

gc()


# _____________________________________________________________
# BOOKINGS_FILTRES_CATEGORIE ####
# Montant total des réservations par deposit_id et catégorie d'offre
# _____________________________________________________________

dbExecute(con, "
  CREATE OR REPLACE TABLE bookings_filtres_categorie AS
  SELECT
    user_id,
    deposit_id,
    offer_category,
    SUM(booking_quantity)          AS nb_bookings,
    SUM(booking_intermediary_amount) AS montant_total
  FROM bookings_filtres_avec_subcategory
  GROUP BY user_id, deposit_id, offer_category
")

dbExecute(con, glue("
  COPY bookings_filtres_categorie
  TO '{chemin_intermediaires}bookings_filtres_categorie.parquet'
  (FORMAT PARQUET)
"))

gc()



# _____________________________________________________________
# DÉCONNEXION ####
# _____________________________________________________________

dbDisconnect(con, shutdown = TRUE)
